package com.hadoop.project.mvc.dao;

import com.hadoop.project.entity.hbase.*;
import com.hadoop.project.gen.maodaligendata.HBaseUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Repository;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @author Alexmaodali
 * 获取所有的hbase数据
 */
@Repository
public class GetAllDao {
    public HBaseResult execute(String tableName) throws Exception {
        // 结果List初始化
        List<Tab> tabList             = new ArrayList<>();
        List<Video> videoList         = new ArrayList<>();
        List<HBaseUser> hBaseUserList = new ArrayList<>();
        List<UserLog> userLogList     = new ArrayList<>();

        // 查询准备工作
        Connection conn           = HBaseUtils.getConnection();
        Table table               = conn.getTable(TableName.valueOf(tableName));
        Scan scan                 = new Scan();
        ResultScanner scanner     = table.getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();

        while (iterator.hasNext()) {
            Result next = iterator.next();

            Tab tab             = new Tab();
            HBaseUser hBaseUser = new HBaseUser();
            Video video         = new Video();
            UserLog userLog     = new UserLog();

            // 获取到单元格数据
            List<Cell> cells = next.listCells();
            for (Cell cell : cells) {
                // System.out.println("rowkey:" + Bytes.toString(CellUtil.cloneRow(cell)));
                // System.out.println("columnFamily:" + Bytes.toString(CellUtil.cloneFamily(cell)));
                // System.out.println("Qualifier:" + Bytes.toString(CellUtil.cloneQualifier(cell)));
                // System.out.println("value:" + Bytes.toString(CellUtil.cloneValue(cell)));
                // System.out.println("--------------------------------------------------------------------------------");

                String rowkey       = Bytes.toString(CellUtil.cloneRow(cell));
                String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
                String column       = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value        = Bytes.toString(CellUtil.cloneValue(cell));

                // 查询Tab
                if ("tab".equals(columnFamily)) {
                    if ("tid".equals(column)) tab.setTid(value);
                    if ("tabName".equals(column)) tab.setTname(value);
                }

                // 查询User
                if ("user".equals(columnFamily)) {
                    if ("uid".equals(column)) hBaseUser.setUid(value);
                    if ("nickname".equals(column)) hBaseUser.setNickName(value);
                    if ("sex".equals(column)) hBaseUser.setSex(value);
                    if ("age".equals(column)) hBaseUser.setAge(value);
                    if ("province".equals(column)) hBaseUser.setProvince(value);
                }

                // 查询Video
                if ("video".equals(columnFamily)) {
                    if ("vid".equals(column)) video.setVid(value);
                    if ("tabs".equals(column)) video.setTabs(value);
                    if ("author".equals(column)) video.setAuthor(value);
                    if ("flip".equals(column)) video.setFlip(value);
                }

                // 查询UserLog
                if ("userlog".equals(columnFamily)) {
                    if ("id".equals(column)) userLog.setId(value);
                    if ("uid".equals(column)) userLog.setUid(value);
                    if ("vid".equals(column)) userLog.setVid(value);
                    if ("videoTime".equals(column)) userLog.setVideoTime(value);
                }
            }

            tabList.add(tab);
            hBaseUserList.add(hBaseUser);
            videoList.add(video);
            userLogList.add(userLog);
        }
        table.close();
        conn.close();

        HBaseResult hBaseResult = new HBaseResult();
        hBaseResult.setTabList(tabList);
        hBaseResult.setHBaseUserList(hBaseUserList);
        hBaseResult.setUserLogList(userLogList);
        hBaseResult.setVideoList(videoList);

        return hBaseResult;
    }
}
